package com.example.listener;

import com.example.lock.Lock;
import com.example.service.SysJobService;
import com.example.service.SysServerService;
import com.example.utils.ListUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.sql.DataSource;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author chentao
 * @version 1.0
 * @description: 实现任务的分配和故障转移
 * @date 2025/8/28 23:31
 */
@Service
public class SysJobTransferListener implements InitializingBean, DisposableBean {

    Logger logger = LoggerFactory.getLogger(getClass());

    @Value("${sys.server.heartbeat.period}")
    private int heartbeatPeriod;

    @Value("${sys.server.heartbeat.count}")
    private int heartbeatCount;

    @Resource
    private DataSource dataSource;

    @Resource
    private SysServerService sysServerService;

    @Resource
    private SysJobService sysJobService;

    private ScheduledExecutorService scheduledExecutorService;

    @Override
    public void afterPropertiesSet() throws Exception {
        this.scheduledExecutorService = Executors.newScheduledThreadPool(10);
        this.scheduledExecutorService.scheduleAtFixedRate(new JobFailedTransferThread(), 0, this.heartbeatPeriod, TimeUnit.SECONDS);
    }

    @Override
    public void destroy() throws Exception {
        this.scheduledExecutorService.shutdownNow();
    }

    /***
     * @author chentao
     * @description 定时任务，每隔period秒执行一次，检查是否有任务需要故障转移
     * @date 2025/8/28 23:32
     */
    class JobFailedTransferThread implements Runnable {

        @Override
        public void run() {
            Lock lock = null;
            try {
                lock = new Lock(dataSource);
                lock.lock();
                List<Long> ids = sysServerService.getDownServerList(heartbeatPeriod * heartbeatCount);
                if (ids.isEmpty()) {
                    List<Long> activeServers = sysServerService.getActiveServerKeys();
                    for (int i = 0; i < activeServers.size(); i++) {
                        Long serverKey = activeServers.get(i);
                        sysJobService.updateJobServerKey(activeServers.size(), i, serverKey);
                    }
                } else {
                    List<Long> activeServers = ListUtils.remove(sysServerService.getActiveServerKeys(), ids);
                    for (Long deadServerKey : ids) {
                        for (int i = 0; i < activeServers.size(); i++) {
                            Long serverKey = activeServers.get(i);
                            sysJobService.updateJobServerKey(activeServers.size(), i, deadServerKey, serverKey);
                        }
                        logger.warn("server key {} is down, transfer jobs to active server key {}", deadServerKey, activeServers);
                    }
                    sysServerService.updateServerStatus(ids);
                }
            } catch (Exception e) {
                logger.warn("jobFailedTransferThread run error: ", e);
            } finally {
                if (lock != null) {
                    lock.unlock();
                }
            }
        }
    }
}
